# Define your item pipelines here
#
# Don't forget to add your pipeline to the ITEM_PIPELINES setting
# See: https://docs.scrapy.org/en/latest/topics/item-pipeline.html


# useful for handling different item types with a single interface
import json

import pymysql
from itemadapter import ItemAdapter
from kafka import KafkaProducer
from scrapy.utils.project import get_project_settings


class NewbilispiderScrapyPipeline:
    def process_item(self, item, spider):
        print(item)
        return item


class MySQLPipeline:
    def __init__(self):
        # 连接数据库
        settings = get_project_settings()
        self.host = settings['DB_HOST']
        self.port = settings['DB_PORT']
        self.user = settings['DB_USER']
        self.password = settings['DB_PASSWORD']
        self.name = settings['DB_NAME']
        self.charset = settings['DB_CHARSET']
        self.connect = pymysql.connect(
            host=self.host,
            port=self.port,
            user=self.user,
            password=self.password,
            db=self.name,
            charset=self.charset,
        )
        # 通过cursor执行增删查改
        self.cursor = self.connect.cursor()

        self.producer = KafkaProducer(value_serializer=lambda v: json.dumps(v).encode('gbk'),
                                      bootstrap_servers=['master"9092'])

    def process_itme_to_kafka(self, item):
        tmp = {
            'bv': item['bv'],
            'username': item['username'],
            'content': item['content'],
            'senddate': item['senddate'],
            'likes': item['likes'],
            'sex': item['sex'],
            'userlevel': item['userlevel'],
        }

        self.producer.send('comment_list', tmp)

    def process_item(self, item, spider):
        self.cursor.execute(
            """insert into hot_comment(bv, username, content, senddate, likes, sex, userlevel)
               value (%s, %s, %s, %s, %s, %s, %s)""",
            (
                item['bv'],
                item['username'],
                item['content'],
                item['senddate'],
                item['likes'],
                item['sex'],
                item['userlevel'],
            )
        )
        self.connect.commit()
        self.process_itme_to_kafka(item)

        # if isinstance(item, SchoolListItem):
        #     self.process_itme_to_kafka(item)
        return item

    def close_spider(self, spider):
        self.cursor.close()
        self.connect.close()
